package com.danbay.framework.kafka;

import com.danbay.framework.redis.RedisClient;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.HashMap;
import java.util.Map;

/**
 * @author SJG
 *         2017/11/13.
 */
@Data
@Slf4j
@ConfigurationProperties
public class KafkaClient<V> extends KafkaTemplate<String, AccessPackage<V>> {

    @Value("${app.kafka.owner.topic-name:unnamed}")
    private String ownerTopicName;
    @Value("${app.kafka.access.topic-name:access}")
    private String accessTopicName;

    @Autowired
    private RedisClient redisClient;

    private Map<String, AccessKafkaTopic> topicMap = new HashMap<>(5);

    public KafkaClient(ProducerFactory<String, AccessPackage<V>> producerFactory) {
        super(producerFactory);
    }

    public KafkaClient(ProducerFactory<String, AccessPackage<V>> producerFactory, boolean autoFlush) {
        super(producerFactory, autoFlush);
    }

    /**
     * 发送一个access的数据包
     *
     * @param accessPackage access数据包
     */
    @Override
    public ListenableFuture<SendResult<String, AccessPackage<V>>> send(String topic, AccessPackage<V> accessPackage) {
        super.send(topic, accessPackage);
        this.flush();
        try {
            log.info("send package:{}", new ObjectMapper().writeValueAsString(accessPackage));
        } catch (JsonProcessingException e) {
            log.error("json writeValueAsString error", e);
        }
        return null;
    }

    /**
     * 新建一个请求消息topic，之后同一个topic的消息都从用此topic发送
     *
     * @param topic       主题名
     * @param messageType 消息类型
     * @return AccessKafkaTopic
     */
    private AccessKafkaTopic newTopic(String topic, byte messageType) {
        //缓存同类型的主题
        String topicKey = topic + messageType;
        if (topicMap.containsKey(topicKey)) {
            return topicMap.get(topicKey);
        }
        AccessKafkaTopic accessKafkaTopic = new AccessKafkaTopic(topic, ownerTopicName, messageType, this, redisClient);
        topicMap.put(topicKey, accessKafkaTopic);
        return accessKafkaTopic;
    }

    /**
     * 新建一个请求消息topic，之后同一个topic的消息都从用此topic发送
     *
     * @param topic 主题名
     * @return AccessKafkaTopic
     */
    public AccessKafkaTopic requestTopic(String topic) {
        return newTopic(topic, AccessPackage.MESSAGE_TYPE_REQUEST);
    }

    /**
     * 新建一个请求消息access topic，之后同一个topic的消息都从用此topic发送
     *
     * @return AccessKafkaTopic
     */
    public AccessKafkaTopic requestAccessTopic() {
        return newTopic(accessTopicName, AccessPackage.MESSAGE_TYPE_REQUEST);
    }

    /**
     * 新建一个请求消息命令处理成功 topic，之后同一个topic的消息都从用此topic发送
     *
     * @return AccessKafkaTopic
     */
    public AccessKafkaTopic requestCmdSuccessTopic() {
        return newTopic(ownerTopicName+"-success", AccessPackage.MESSAGE_TYPE_REQUEST);
    }

    /**
     * 新建一个响应消息topic，之后同一个topic的消息都从用此topic发送
     *
     * @param topic 主题名
     * @return AccessKafkaTopic
     */
    public AccessKafkaTopic responseTopic(String topic) {
        return newTopic(topic, AccessPackage.MESSAGE_TYPE_RESPONSE);
    }
}
